package java8.completablefuture;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @program: util
 * @description:
 * @author: zhouliangliang
 * @create: 2019-09-26 10:35
 **/
public class CompletableFutureTest {

    private static Random random = new Random();

    List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
            new Shop("LetsSaveBig"),
            new Shop("MyFavoriteShop"),
            new Shop("BuyItAll"));

    private final Executor executor = Executors.newFixedThreadPool(Math.min(shops.size(), Runtime.getRuntime().availableProcessors()),
            new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread();
                    thread.setDaemon(true);
                    return thread;
                }
            });



    public static void delay(long seconds) {
        try {
            Thread.sleep(seconds);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public static void randomDelay() {
        int delay = 500 + random.nextInt(2000);
        delay(Long.valueOf(delay));
    }


    private double calculatePrice(String product) {
        delay(1000l);
        return new Random().nextDouble() * product.charAt(0) + product.charAt(1);
    }

    //同步
    public double getPriceBlock(String product) {
        return calculatePrice(product);
    }

    //异步
    public Future<Double> getPriceAsync(String product) {
        java.util.concurrent.CompletableFuture<Double> future = new java.util.concurrent.CompletableFuture<>();
        new Thread(()->{
            double price = calculatePrice(product);
            future.complete(price);
        }).start();
        return future;
    }

    //异步处理异常
    public Future<Double> getPriceAsyncExcep(String product) {
        java.util.concurrent.CompletableFuture<Double> future = new java.util.concurrent.CompletableFuture<>();
        new Thread(()->{
            try{
                double price = calculatePrice(product);
                future.complete(price);
            } catch (Exception ex) {
                future.completeExceptionally(ex);
            }
        }).start();
        return future;
    }

    //labma
    public Future<Double> getPriceAsyncLamba(String product) {
        return java.util.concurrent.CompletableFuture.supplyAsync(()->calculatePrice(product));
    }

    public List<String> findPrice(String product) {
        // stream
//        return shops.stream().map(shop ->String.format("%s price is %.2f", shop.getName(), shop.getPrice(product))).collect(Collectors.toList());
        // parallel stream

        return shops.parallelStream().map(shop ->String.format("%s price is %.2f", shop.getName(), shop.getPrice(product))).collect(Collectors.toList());


    }
    /**
     * CompletableFuture 可以指定线程池executor
     * @return
     */
    public List<String> findPrice2(String product) {
        List<CompletableFuture<String>> result = shops
                .stream()
                .map(shop ->
            CompletableFuture.supplyAsync(() ->String.format("%s price is %.2f", shop.getName(), shop.getPrice(product)), executor)).collect(Collectors.toList());
        //join 操作，可以获得每一个CompletableFuture的结果
        return result.stream().map(CompletableFuture::join).collect(Collectors.toList());
    }

    public String getPrice3(String product) {
        double price = calculatePrice(product);

        //随机选取一个code
        Discount.Code code = Discount.Code.values()[
                random.nextInt(Discount.Code.values().length)];

        return String.format("%s:%.2f:%s", product, price, code);
    }


    private List<String> findPrice3(String product) {
        return shops.stream()
                .map(shop -> getPrice3(product))
                .map(Quote::parse)
                .map(Discount::applyDiscount)
                .collect(Collectors.toList());
    }

    /**
     * thenCompose方法允许你对两个异步操作进行流水线，第一个操作完成时，将其 结果作为参数传递给第二个操作
     * thenCombine方法，它接收名为BiFunction的第二参数，这个参数 定义了当两个CompletableFuture对象完成计算后，结果如何合并
     * @param product
     */
    private void findPrice4(String product) {
        List<CompletableFuture<String>> priceFuture = shops.stream()
                .map(shop -> CompletableFuture.supplyAsync(() -> getPrice3(product))) //一个异步线程获取价格
                 .map(future ->future.thenApply(Quote::parse)) //对返回值进行转换
                 .map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote), executor))) //另一个异步线程获取折扣
                 .collect(Collectors.toList());

        priceFuture.stream().map(CompletableFuture::join)//流中的所有future执行完毕，提取返回值 join相当于get
        .collect(Collectors.toList());

    }

    public Stream<CompletableFuture<String>> findPricesStream(String product) {
        return shops.stream()
                .map(shop -> CompletableFuture.supplyAsync(() ->getPrice3(product), executor))
                .map(future -> future.thenApply(Quote::parse))
                .map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync(() ->Discount.applyDiscount(quote), executor)));
    }

    private void findPrice5(String product) {
        CompletableFuture[] futures = findPricesStream(product).map(f -> f.thenAccept(System.out::println))
                .toArray(size -> new CompletableFuture[size]);
        CompletableFuture.allOf(futures).join();
    }

    public static void main(String args[]) {

        long start = System.nanoTime();
//        System.out.println(new CompletableFutureTest().findPrice5("myPhone27S"));
//        new CompletableFutureTest().findPrice5("myPhone27S");

        CompletableFuture[] futures = new CompletableFutureTest().findPricesStream("myPhone27S")
                .map(f ->f.thenAccept(s -> System.out.println(s + "(done in" + ((System.nanoTime() - start)/1000000) + " msecs")))
                .toArray(size -> new CompletableFuture[size]);
        CompletableFuture.allOf(futures).join();

        System.out.println("All shops have now responded in "
                + ((System.nanoTime() - start) / 1_000_000) + " msecs");

//        long duration = (System.nanoTime() - start) / 1_000_000;
//        System.out.println("Done in " + duration + " msecs");


    }

    class Shop {

        private String name;
        private Double price;

        public Shop(String name) {
            this.name = name;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public Double getPrice(String product) {
            return getPriceBlock(product);
        }
    }
}
